package scala_hw6
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf



//1.创建一个包含100个元素的RDD，将其中大于50的元素乘以2，并筛选出其中大于100的元素。
object first {
  def main(args: Array[String]): Unit = {
    val rdd = sc.parallelize(1 to 100)
    val rdd2 = rdd.filter(_ > 50).map(_ * 2).filter(_ > 100)
    val result = rdd2.collect()
    println(result.mkString(", "))
  }
}

//2.创建一个包含10个元素的RDD，将其中所有元素按空格切分成单词，并将所有单词变成小写形式（转换小写toLowerCase）。
object second {
  def main(args: Array[String]): Unit = {
    val rdd = sc.parallelize(Seq("Hello World", "Spark is great", "Scala is awesome"))
    val rdd2 = rdd.flatMap(_.toLowerCase.split(" "))
    val result = rdd2.collect()
    println(result.mkString(", "))

  }
}


//3.创建一个包含5个元素的RDD，每个元素都是一个二元组，第一个元素是字符串类型的名字，第二个元素是Int类型的分数。例如：Seq(("Alice", 90), ("Bob", 80), ("Charlie", 70), ("Alice", 85), ("Bob", 95))
//计算每个名字对应的分数之和。
object third {
  def main(args: Array[String]): Unit = {
    val rdd = sc.parallelize(Seq(("Alice", 90), ("Bob", 80), ("Charlie", 70), ("Alice", 85), ("Bob", 95)))
    val rdd2 = rdd.reduceByKey(_ + _)
    val result = rdd2.collect()
  }
}



object four {
  def main(args: Array[String]): Unit = {
//    4.创建一个包含一些字符串的RDD，例如：
    val rdd = sc.parallelize(Seq("apple", "banana", "orange", "pear", "watermelon", "grape", "pineapple"))
//    现在你需要按照字符串长度将其分为两个分区，其中一个分区包含长度小于等于5的字符串，另一个分区包含长度大于5的字符串。

    // 创建一个包含一些字符串的RDD
    val rdd = sc.parallelize(Seq("apple", "banana", "orange", "pear", "watermelon", "grape", "pineapple"))

    // 定义一个自定义分区器，将长度小于等于5的字符串分到第一个分区，将长度大于5的字符串分到第二个分区
    class CustomPartitioner extends Partitioner {
      // 定义分区数量为2
      override def numPartitions: Int = 2
      // 返回字符串所属的分区编号
      override def getPartition(key: Any): Int = {
        key match {
          case str: String =>
            // 如果字符串长度小于等于5，则分到第一个分区
            if (str.length <= 5) 0 else 1
          case _ => throw new IllegalArgumentException("Expected a string")
        }
      }
    }

    // 对RDD使用自定义分区器进行分区，并进行持久化操作
    val partitionedRDD = rdd.map(str => (str, 1)).partitionBy(new CustomPartitioner).persist()

    // 对每个分区内的字符串进行计数，并打印计数结果
    // 首先对每个分区进行计数，并返回一个迭代器
    val resultRDD = partitionedRDD.mapPartitions(iter => Iterator(iter.size))
    // 打印计数结果
    resultRDD.collect().foreach(println)
  }
}








